package com.gome.boot.config.elasticsearch.cluster;

import com.gome.boot.common.enums.ExecutorTypeEnum;
import com.gome.boot.common.logger.BaseLogger;
import com.gome.boot.common.util.ScheduleUtils;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.springframework.data.elasticsearch.ElasticsearchException;

import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * ES集群监听器.
 *
 * @author baoxiufeng
 */
public class ClusterMonitor {

    /** 单例对象 */
    private volatile static ClusterMonitor instance = null;
    /** 监听线程 */
    private ScheduledExecutorService executorService = null;

    /** 集群监听时间频率 */
    private static final int MONITOR_PERIOD = 60;

    /**
     * 私有构造方法.
     */
    private ClusterMonitor() {
        executorService = ScheduleUtils.get().getExecutor(ExecutorTypeEnum.SINGLE_SCHED);
    }

    /**
     * 获取集群监听器实例.
     *
     * @return 集群监听器实例
     */
    public static ClusterMonitor getInstance() {
        if (instance == null) {
            synchronized (ClusterMonitor.class) {
                if (instance == null) {
                    instance = new ClusterMonitor();
                }
            }
        }
        return instance;
    }

    /**
     * 启动集群监听器
     */
    public void start(MultiClusterClient multiClusterClient) {
        executorService.scheduleWithFixedDelay(new ClusterMonitorThread(multiClusterClient), MONITOR_PERIOD, MONITOR_PERIOD, TimeUnit.SECONDS);
    }

    /**
     * 集群监听线程.
     *
     * @author baoxiufeng
     */
    private class ClusterMonitorThread extends BaseLogger implements Runnable {

        /**
         * ES多集群客户端
         */
        private MultiClusterClient multiClusterClient;

        /**
         * ES集群客户端集合
         */
        private Map<String, ClusterClient> clientMap;

        /**
         * 集群健康状态查询请求实例.
         */
        private final ClusterHealthRequest healthRequest = new ClusterHealthRequest();

        /**
         * 构造方法.
         *
         * @param multiClusterClient ES多集群客户端
         */
        public ClusterMonitorThread(MultiClusterClient multiClusterClient) {
            this.multiClusterClient = multiClusterClient;
            this.clientMap = multiClusterClient.getClientMap();
        }

        @Override
        public void run() {
            logger.info("Start to check cluster stats...");
            String clusterName = null;
            ClusterClient clusterClient = null;
            int currAvailableClusterCount = 0;
            for (Map.Entry<String, ClusterClient> entry : clientMap.entrySet()) {
                clusterName = entry.getKey();
                clusterClient = entry.getValue();
                ClusterHealthStatus lastStatus = clusterClient.getHealthStats();
                ClusterHealthStatus currStatus = ClusterHealthStatus.RED;
                try {
                    currStatus = clusterClient.getClient().admin().cluster().health(healthRequest).actionGet().getStatus();
                } catch (ElasticsearchException e) {
                    logger.error("Execute cluster[" + clusterName + "] health request failure.", e);
                }
                logger.info("Cluster[{}] current status is '{}'", clusterName, currStatus);
                if (lastStatus != currStatus) {
                    clusterClient.setHealthStats(currStatus);
                    if (ClusterHealthStatus.RED == currStatus) {
                        logger.warn("Cluster[{}] is unavailable now, change state from '{}' to '{}'", clusterName, lastStatus, currStatus);
                        clusterClient.setStats(ClusterStats.UNAVAILABLE);
                    }
                }
                if (ClusterStats.AVAILABLE == clusterClient.getStats()) {
                    currAvailableClusterCount++;
                }
            }
            if (currAvailableClusterCount == 0) {
                logger.warn("!!!!!! Now have not any available cluster !!!!!!");
            } else {
                logger.info("Current available cluster count:[{}]", currAvailableClusterCount);
            }
        }
    }
}
